#include "app_pool.h"
#include "mqueue.h"
#include "unistd.h"
#include "pthread.h"
#include "stdio.h"
#include "fcntl.h"
#include "log.h"

static mqd_t mqd;
static pthread_t pool_threads[POOL_THREAD_SIZE];

void *app_pool_handle_task(void *arg)
{
    while (1)
    {
        app_pool_task_t task;
        mq_receive(mqd, (char *)&task, sizeof(app_pool_task_t), NULL);
        task.func(task.arg);
    }
}

int app_pool_init(void)
{
    /* 清理掉旧的队列 */
    mq_unlink(POOL_QUEUE_NAME);
    /* 1. 创建消息队列 */
    struct mq_attr attr; 
    attr.mq_maxmsg = 10;   
    attr.mq_msgsize = sizeof(app_pool_task_t);

    mqd = mq_open(POOL_QUEUE_NAME, O_RDWR | O_CREAT, 0644, &attr);
    if (mqd == -1)
    {
        perror("mq_open failed");
        return -1;
    }
    
    /* 2. 创建线程池 */
    for (size_t i = 0; i < POOL_THREAD_SIZE; i++)
    {
        pthread_create(&pool_threads[i], NULL, app_pool_handle_task, NULL);
    }
    return 0;
}


int app_pool_submit_task(void *(*func)(void *), void *arg)
{
    /* 1.把函数和参数封装成任务  --> 任务结构体 */
    app_pool_task_t task;
    task.func = func;
    task.arg = arg;
    /* 2.把任务放入消息队列中 */
    int ret = mq_send(mqd, (char *)&task, sizeof(app_pool_task_t), 0);
    if (ret == 0)
    {
        perror("mq_send succcess");
        return 0;
    }else{
        perror("mq_send failed");
        return -1;
    }
}


int app_pool_destroy(void)
{
    /* 把线程逐个 cancel */
    for (size_t i = 0; i < POOL_THREAD_SIZE; i++)
    {
        pthread_cancel(pool_threads[i]);
    }

    /* 销毁消息队列 */
    mq_close(mqd);

    /* 删除队列 */
    mq_unlink(POOL_QUEUE_NAME);
}